package com.fwmagic.flink.projectcase.queryactivitycase.sink;

import com.fwmagic.flink.projectcase.queryactivitycase.bean.ActivityBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * kafka->AsyncHttpClient->mysql
 *
 * kafka数据：
 * u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
 * u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
 * u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
 * u003,A3,2019-06-08 10:11:11,3,121.26757,37.49794
 * u003,A3,2019-06-08 10:11:11,3,121.26757,37.49794
 * u002,A2,2019-08-03 10:11:11,2,123.818817,41.312458
 *
 * mysql表
 * CREATE TABLE `t_activity_counts` (
 *   `aid` varchar(10) NOT NULL,
 *   `event_type` int(11) NOT NULL,
 *   `counts` int(11) DEFAULT NULL,
 *   PRIMARY KEY (`aid`,`event_type`)
 * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
 *
 *
 */
public class MySqlSink extends RichSinkFunction<ActivityBean> {

    private transient Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/flink-test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
    }

    @Override
    public void invoke(ActivityBean bean, Context context) throws Exception{
        String sql ="insert into t_activity_counts(`aid`,`event_type`,`counts`) values(?,?,?) ON DUPLICATE KEY UPDATE `counts`=?";
        PreparedStatement ps = null;
        try {
            ps = connection.prepareStatement(sql);
            ps.setString(1, bean.aid);
            ps.setInt(2, bean.eventType);
            ps.setInt(3, bean.counts);
            ps.setInt(4, bean.counts);
            ps.executeUpdate();
        } finally {
            if(ps!=null) ps.close();
        }
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
